package com.gxljc.bear.crawler.weixin;

import com.mongodb.BasicDBObject;
import com.mongodb.Bytes;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.gxljc.commons.mongo.cli.BaseMongoRepository;
import com.gxljc.commons.util.Pair;
import com.gxljc.bear.crawler.base.BaseMongoTools;
import com.gxljc.bear.crawler.base.DolphinCrawlerConsts;
import com.gxljc.bear.crawler.base.DolphinFetchData;
import com.gxljc.bear.crawler.image.ImageFresh;
import com.gxljc.bear.crawler.proxy.ProxyUtil;
import com.gxljc.bear.crawler.util.HtmlUtil;
import com.gxljc.bear.crawler.util.SparkUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

import java.io.Serializable;
import java.util.*;

/**
 * 搜狗微信PAGE数据爬虫。
 *
 * @author tanghaitao
 * @since 2022-8-3
 */
public class PcPageCrawler implements Serializable {
    public static Logger LOG = Logger.getLogger(PcPageCrawler.class);
    private Boolean proxy = false;

    public PcPageCrawler(Boolean proxy) {
        this.proxy = proxy;
    }

    public void crawl() throws Exception {

        final long SLEEP_TIME = 13 * 60 * 1000;
        while (true) {
            List<PageTable> seeds = null;
            try{
                seeds = getSeed();
            }catch(Exception er){
                er.printStackTrace();
            }

            if (CollectionUtils.isEmpty(seeds)) {
                LOG.error("page is empty, sleep 13 min");
                Thread.sleep(SLEEP_TIME); //没有数据，暂停5分钟
                continue;
            }
            int size = seeds.size();
            ProxyUtil.initQueue(proxy);
            LOG.info("weixin page seeds size = " + size);
            if (size < WeixinConst.MAX_SINGLE_CORE) { //少于一定数量时，使用单线程爬取
                try {
                    crawlNoSpark(seeds);
                } catch (Exception er) {
                    er.printStackTrace();
                    LOG.error("single core job error", er);
                }
            } else {
                try {
                    crawl(seeds);
                } catch (Exception er) {
                    er.printStackTrace();
                    LOG.error("spark job error", er);
                }
            }
            LOG.info("sleep 13 min");
            Thread.sleep(SLEEP_TIME); //5分钟检查一次
        }
    }

    public void crawl(List<PageTable> seeds) throws Exception {
        int coreMax = 15;
        JavaSparkContext jsc = SparkUtil.createCommonsSparkContext(
                "bear-手机微信文章爬虫-" + seeds.size(), coreMax,
                coreMax * 2, PcPageCrawler.class);
        JavaRDD<PageTable> seedsRDD = jsc.parallelize(new ArrayList<PageTable>(seeds));
        long count = seedsRDD.mapPartitions(
                new FlatMapFunction<Iterator<PageTable>, Integer>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterable<Integer> call(Iterator<PageTable> seeds)
                            throws Exception {
                        int successCnt = 0;
                        while (seeds.hasNext()) {
                            PageTable seed = seeds.next();
                            try {
                                int ret = crawl(seed);
                                if (ret == 1)
                                    successCnt++;
                                if (successCnt % 100 == 0) {
                                    LOG.info("successCnt=" + successCnt);
                                }
                            } catch (Exception er) {
                                WeixinUtil.failTime(seed, WeixinConst.WEIXIN_MOBILE_PAGE);
                                er.printStackTrace();
                            }
                        }
                        return Arrays.asList(successCnt);
                    }
                }).count();
        LOG.info("all count=" + count);
        jsc.stop();
    }

    public void crawlNoSpark(List<PageTable> seeds) {
        if (CollectionUtils.isEmpty(seeds)) {
            LOG.error("page is empty");
            return;
        }
        for (PageTable table : seeds) {
            try {
                crawl(table);
            } catch (Exception e) {
                WeixinUtil.failTime(table, WeixinConst.WEIXIN_MOBILE_PAGE);
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
        }
    }

    //单条url入口
    public int crawl(PageTable table) throws Exception {
        LOG.info("crawl = " + table.getCrawlUrl());
        String url = table.getCrawlUrl();
        DolphinFetchData fetchData = DolphinFetchData.getInstance(DolphinCrawlerConsts.CrawlerChannelType.MOVIE.getName());
        byte[] htmlByte = fetchData.getHtml(url, proxy, WeixinConst.CRAWL_TIMEOUT);
        if (htmlByte == null) {
            throw new Exception("page result is empty url = " + url);
        }
        String html = new String(htmlByte);
        if (StringUtils.isEmpty(html)) {
            throw new Exception("page result is empty url = " + url);
        }
        extract(table, html);
        String content = table.getContent();
        Date publishDate = table.publishDate;
        if (StringUtils.isEmpty(content) || publishDate == null)
            throw new Exception("page content,publishDate is error url = " + url);
        LOG.info(table);
        WeixinMongodbUtil.savePageMongodb(table, WeixinConst.WEIXIN_MOBILE_PAGE);
        return DolphinCrawlerConsts.RESULT_YES;
    }

    //抽取原创
    private void extractOriginal(PageTable table, Document doc) {
        Elements dateElement = doc.select("#copyright_logo");
        if (CollectionUtils.isEmpty(dateElement))
            table.setOriginal(0);
        else
            table.setOriginal(1);
    }

    //抽取发布时间
    private void extractPublishDate(PageTable table, String html) {
        String startStr = "var ct = \"";
        String end = "\";";
        if (html.indexOf(startStr) < 0 || html.indexOf(end) < 0) {
            return;
        }
        int first = html.indexOf(startStr) + startStr.length();
        String link = html.substring(first, html.indexOf(end, first));
        if (StringUtils.isEmpty(link)) return;
        table.setPublishDate(new Date(Long.parseLong(link) * 1000));
    }

    //内容抽取
    private void extractContent(PageTable table, Document doc) {
        Elements contentElement = doc.select("#js_content");
        if (CollectionUtils.isEmpty(contentElement)) return;
        String content = contentElement.text();
        if (StringUtils.isEmpty(content)) return;
        table.setContent(content);
    }

    //图片替换
    private void parseHtmlAndImage(PageTable table, Document doc) {
        Elements contentElement = doc.select("#js_content");
        if (CollectionUtils.isEmpty(contentElement)) return;
        String html = contentElement.html();
        table.setHtml(html);
        Elements imageElements = contentElement.select("img");
        if (CollectionUtils.isEmpty(imageElements)) return;
        ImageFresh fresh = ImageFresh.getInstance();
        for (Element imageElement : imageElements) {
            String src = imageElement.attr("src");
            String dataSrc = imageElement.attr("data-src");
            if (StringUtils.isEmpty(src) && StringUtils.isEmpty(dataSrc)) continue;
            String newSrc = src;
            if (!StringUtils.isEmpty(dataSrc)) {
                newSrc = dataSrc;
            }
            String newUrl = HtmlUtil.pictureUrlPref(newSrc, table.crawlUrl);
            Pair<Integer, String> newPicPair = null;
            try {
                newPicPair = fresh.freshOne(newUrl);
                String newPic = newPicPair.second;
                html = html.replace(newUrl, newPic);
            } catch (Exception e) {
                e.printStackTrace();
                continue;
            }
        }
        table.setHtml(html);
    }

    //页面抽取并返回  object
    private void extract(PageTable table, String html) throws Exception {
        Document doc = Jsoup.parse(html);
        extractPublishDate(table, html);
        extractContent(table, doc);
        parseHtmlAndImage(table, doc);
        extractOriginal(table, doc);
        table.setWid(WeixinUtil.genWid(table));
        table.setCrawlFlag(DolphinCrawlerConsts.RESULT_YES);
        table.setRsync(0);
    }

    //获取seed
    public List<PageTable> getSeed() throws Exception {
        BaseMongoRepository repo = BaseMongoTools.getInstance(DolphinCrawlerConsts.MongoDBName.WEIXIN.getName());
        BasicDBObject keys = new BasicDBObject();
        keys.put(DolphinCrawlerConsts.MONGODB_DEFAULT_ID, 1);
        keys.put(WeixinConst.SCHEMA_COLUMN_NAME.CRAWL_URL.getValue(), 1);
        keys.put(WeixinConst.SCHEMA_COLUMN_NAME.ACCOUNT_ID.getValue(), 1);
        keys.put(WeixinConst.SCHEMA_COLUMN_NAME.ACCOUNT_NAME.getValue(), 1);
        keys.put(WeixinConst.SCHEMA_COLUMN_NAME.ORG_URL.getValue(), 1);
        keys.put(WeixinConst.SCHEMA_COLUMN_NAME.TITLE.getValue(), 1);
        keys.put(WeixinConst.SCHEMA_COLUMN_NAME.FAIL_TIME.getValue(), 1);
        DBObject query = new BasicDBObject();
        query.put(WeixinConst.SCHEMA_COLUMN_NAME.CRAWL_FLAG.getValue(), new BasicDBObject("$ne", 1));
        DBCursor cursor = repo.getCollection(WeixinConst.WEIXIN_MOBILE_PAGE)
                .find(query, keys)
                .addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        List<PageTable> seeds = new LinkedList<>();
        List<DBObject> dataIterator = cursor.toArray();
        for (DBObject data : dataIterator) {
            try {
                Object id = data.get(DolphinCrawlerConsts.MONGODB_DEFAULT_ID);
                Object url = data.get(WeixinConst.SCHEMA_COLUMN_NAME.CRAWL_URL.getValue());
                Object orgUrl = data.get(WeixinConst.SCHEMA_COLUMN_NAME.ORG_URL.getValue());
                Object accountId = data.get(WeixinConst.SCHEMA_COLUMN_NAME.ACCOUNT_ID.getValue());
                Object accountName = data.get(WeixinConst.SCHEMA_COLUMN_NAME.ACCOUNT_NAME.getValue());
                Object failTime = data.get(WeixinConst.SCHEMA_COLUMN_NAME.FAIL_TIME.getValue());
                Object title = data.get(WeixinConst.SCHEMA_COLUMN_NAME.TITLE.getValue());
                PageTable table = new PageTable();
                table.setPageId(id.toString());
                table.setAccountId(accountId.toString());
                table.setAccountName(accountName.toString());
                table.setCrawlUrl(url.toString());
                table.setOrgUrl(orgUrl.toString());
                table.setTitle(title.toString());
                if (failTime != null) table.setFailTime((Integer) failTime);
                seeds.add(table);
            } catch (Exception er) {
                er.printStackTrace();
            }
        }
        return seeds;
    }
}
